package com.redis.demo.rabbitmq.Publisher;

import com.rabbitmq.client.*;
import com.redis.demo.rabbitmq.Constants.Constants;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

public class PublisherConfirms {
    private static final Integer MESSAGE_COUNT=10000 ;
    static Connection createConnection() throws IOException, TimeoutException {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost(Constants.HOST);
        factory.setPort(Constants.PORT);
        factory.setUsername(Constants.USERNAME);
        factory.setPassword(Constants.PASSWORD);
        //设置虚拟机的地址
        factory.setVirtualHost(Constants.VIRTUALHOST);
        Connection connection=factory.newConnection();
        return connection;
    }
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        //单独确认
        //publishingMessageIndividually();
        //批量确认
        publishingMessageInBatches();
        //异步确认
        handlingPublisherConfirmsAsynchronously();
    }

    private static void handlingPublisherConfirmsAsynchronously() throws IOException, TimeoutException, InterruptedException {
        //异步确认
        try(Connection connection=createConnection()) {
            Channel channel = connection.createChannel();
            channel.confirmSelect();
            long start=System.currentTimeMillis();
            SortedSet<Long> confirmSet=Collections.synchronizedSortedSet(new TreeSet<>());
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    if (multiple){
                        //将小于deliveryTag的值删除
                        confirmSet.headSet(deliveryTag+1).clear();
                    }else {
                        confirmSet.remove(deliveryTag);
                    }
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    //nack需要重新发送
                    if (multiple){
                        //将小于deliveryTag的值删除
                        confirmSet.headSet(deliveryTag+1).clear();
                    }else {
                        confirmSet.remove(deliveryTag);
                    }
                    //需要处理，比如重发消息



                }
            });

            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg="hello publisher confirm"+i;

                //指定id
                long seqNo=channel.getNextPublishSeqNo();
                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE3,null,msg.getBytes());
                //在集合中放入序号
                confirmSet.add(seqNo);


            }
            while (!confirmSet.isEmpty()){
                Thread.sleep(10);
            }

            long end=System.currentTimeMillis();
            System.out.println("异步确认" +( end-start));



        }


    }

    private static void publishingMessageInBatches() throws IOException, TimeoutException, InterruptedException {
        try(Connection connection=createConnection()) {
            Channel channel = connection.createChannel();
            channel.confirmSelect();
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE2,true,false,false,null);
            //设置批量数为100
            int batchSize=100;
            int outstandingMessageCount=0;
            Long start=System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                outstandingMessageCount++;
                String msg="hello  publishingMessageInBatches"+i;
                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE2,null,msg.getBytes());
                if (outstandingMessageCount==batchSize){
                    channel.waitForConfirmsOrDie(5000);
                    outstandingMessageCount=0;
                }


            }
            if (outstandingMessageCount>0){
                channel.waitForConfirmsOrDie(5000);
            }

            Long end=System.currentTimeMillis();
            System.out.println("批量确认"+(end-start));


        }



    }

    private static void publishingMessageIndividually() throws IOException, TimeoutException, InterruptedException {
        try (Connection connection=createConnection()){

            Channel channel = connection.createChannel();
            //将信道设置为确认模式
            channel.confirmSelect();

            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE1,true,false,false,null);
            Long start=System.currentTimeMillis();
            for (int i = 0; i <MESSAGE_COUNT ; i++) {
                String msg="hello publishingMessageIndividually"+i;
                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE1,null,msg.getBytes());
                //等待mq存储到硬盘上才会返回，这种策略比较慢
                channel.waitForConfirmsOrDie(5000);
            }
            Long end=System.currentTimeMillis();
            System.out.println("立即确认"+(end-start));

        }
    }



}
